[Amazon SQS] FIFOキューを使用したメール送信（VUIのバックエンドのレスポンスを高速化する）

SIN

2019.10.23

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX事業本部の平内（SIN）です。

Amazon Connectや、Alexaなど(※1)、VUIなシステムで利用されるLambdaでは、UXへの配慮のためか、タイムアウトが厳密に定義されており、どちらも、レスポンスが8秒以上かかると、エラーとして処理されてします。

このような制限の中で、活用されるのが、Amazon SQS(以下、SQS)です。SQSで、処理を階層（非同期）化して簡単にボトルネックを解消できます。また、フルマネージドなサービスであるSQSを挟むことで、耐障害性の向上も同時に図れます。

なお、昨年末より東京リージョンも利用可能になっている、FIFOキューでは、順番や、重複の制御も必要ないので、要件にもよりますが、軽易にキューを扱えると言えます。

今回は、VUIなバックエンドでメールを送ることを想定しFIFOキューで処理する要領を纏めます。

記事の内容は、単純なSQS（FIFO)の実装要領です。自分用の覚書であることをお許しください。

※1 Amazon Lexは、VUIとは限らない性質も有り、タイムアウト値（セッションタイムアウト）は、ボット作成時に作成者が分単位で自由に設定できます。

2 考慮すべき制限（要件）等

念のため、SQS（FIFO）を使用する場合に、考慮すべき制限（要件）等を列挙しておきます。
参考：Amazon Simple Queue Service とは

処理速度は、バッチ処理でない場合、300件/秒 (1秒あたりの送信、受信、又は削除のオペレーション数)
グループIDが同じ場合、重複するメッセージは、自動的に削除される（削除されたくない場合は、メッセージごとに違うグループIDを付与する）
グループIDが同じ場合、順序が保たれる（順次処理したくない場合は、メッセージごとに違うグループIDを付与する）
メッセージの重複排除には、メッセージ重複排除IDを指定する
キュー作成時にコンテンツに基づく重複排除にチェックすることで、自動的にコンテンツハッシュ値（SHA256）が生成され、上記の、メッセージ重複削除IDは必要なくなる

シーケンス番号は、SQSによって各メッセージに割り当てられる連続番号
メッセージの削除は、可視性タイムアウト内に行う必要がある
メッセージサイズは、256KB以内

3 メール送信のLambda

サンプルに使用するのは、Twilioでショートメールを送るコードです。メール送信の本体は、sendmail()です。Performance　クラスは、単純に計測するためのものです。

import * as AWS from 'aws-sdk';
import { performance } from 'perf_hooks';

exports.handler = async (event: any) => {

    const phoneNumber = process.env.PHONE_NUMBER;
    const message = 'テストメッセージ';

    const performance = new Performance("メール送信"); // 計測開始
    const response = await sendmail(phoneNumber, message); // メール送信
    performance.measurement(); // 計測終了

    console.log(response);
}

async function sendmail(phoneNumber: string, message: string) {
    const account_sid = process.env.TWILIO_ACCOUNT_SID;
    const auth_token = process.env.TWILIO_AUTH_TOKEN;
    const from = process.env.TWILIO_NUMBER;

    const client = require('twilio')(
        account_sid,
        auth_token
    );
    return new Promise((resolve,reject) => {
        client.messages
        .create({
            body: message,
            from: from,
            to: phoneNumber
        })
        .then((result: any) => {
            resolve(result);
        })
        .catch( (err: any) => {
            reject(err);
        });
    })
}

export class Performance {
    startTime = 0;
    title: string ='';

    constructor(title:string){
        this.startTime = performance.now();
        this.title = title;
    }

    measurement(){
        const span = performance.now() - this.startTime;
        console.log(`[performance] ${this.title} ${(span/1000).toFixed(2)}sec`);
    }
}

sendmail()の前後で計測して、2.01secとなってました。（※ 処理時間は、条件によって大きく変化します）

[performance] メール送信 2.01sec

4 SQS

特別な設定無しで、FIFOのキュー（sample.fifo）を作成しました。

4 プロデューサー

sendmail()をコンシューマー側に移動し、プロデューサーになるようにsendmail()の内容をSQSへの送信に書き換えます。

async function sendmail(phoneNumber: string, message: string) {

  const body = {
    phoneNumber: phoneNumber,
    message: message
  }

  const account  = process.env.ACCOUNT;
  const region = 'ap-northeast-1';
  const queueName = 'sample.fifo';
  const url = `https://sqs.${region}.amazonaws.com/${account}/${queueName}`;
  const deduplicationId = Math.random().toString(32).substring(2); // 重複制御
  const groupId = 'sqs-fifo-sample'; // 同じグループとする

  const sqs = new AWS.SQS();
  const params: AWS.SQS.Types.SendMessageRequest = {
    MessageBody: JSON.stringify(body),
    MessageGroupId: groupId,
    MessageDeduplicationId: deduplicationId,
    QueueUrl: url,
  };
  return await sqs.sendMessage(params).promise();
}

Twilioへ送信をSQSへの送信に切り替えたことで，一応処理時間は短縮されています。

[performance] メール送信 0.37sec

5 コンシューマー

コンシューマー側は、以下のようになります。SQSから受信したキューの内容でメールを送信します。キューは、処理した時点で、削除しています。可視性タイムアウトは、明示的に指定していないので、デフォルトの30秒となっています。

import * as AWS from 'aws-sdk';

exports.handler= async (event: any) => {

  const sqs = new AWS.SQS();
  const account  = process.env.ACCOUNT;
  const region = 'ap-northeast-1';
  const queueName = 'sample.fifo';
  const url = `https://sqs.${region}.amazonaws.com/${account}/${queueName}`;

  var params: AWS.SQS.ReceiveMessageRequest = {
    MaxNumberOfMessages: 10,
    QueueUrl: url,
    WaitTimeSeconds: 0
   };

   await sqs.receiveMessage(params, (err, data) => {
    if (err) {
      console.log("Receive Error", err);
    } else if (data.Messages) {
      for(var i=0; i<data.Messages.length; i++) {
        const message = data.Messages[i];
        const body:{phoneNumber: string, message: string} = JSON.parse(message.Body!)
        // メール送信
        await sendmail(body.phoneNumber, body.message)
        var deleteParams = {
          QueueUrl: url,
          ReceiptHandle: data.Messages[i].ReceiptHandle!
        };
        sqs.deleteMessage(deleteParams, function(err, data) {
          if (err) {
            console.log("Delete Error", err);
          } else {
            console.log("Message Deleted", data);
          }
        });
      }
    }
  });
}